package base.java9_17;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Processor;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;

/**
 * basic.java9 reactive stream 响应式流实现
 * <p>
 * * {@link Flow}
 * * {@link SubmissionPublisher}
 * * {@link Flow.Subscriber}
 * * {@link Processor}
 *
 * @author 王霄
 * @date 2021/9/26 13:38
 **/
public class FlowDemo {

    private static final ExecutorService executor = Executors.newSingleThreadExecutor();

    private static final int maxBufferCapacity = 16;

    public static void main(String[] args) {
        //创建发布者
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>(executor, maxBufferCapacity);

        //创建订阅者
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println("订阅器处理开始");
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                subscription.request(1);
                System.out.println("消费：" + item);
            }

            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
                subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("订阅器处理完成");
            }
        };

        //建立订阅关系
        publisher.subscribe(subscriber);

        //发布消息
        IntStream.rangeClosed(1, 50).forEach(publisher::submit);

        //关闭发布者
        publisher.close();

        //关闭线程池
        executor.shutdownNow();
    }
}
